package com.bjtu.boatmq.client.consumer;

import cn.boat.router.consumer.ConsumerMetaData;
import cn.boat.router.consumer.GetConsumerMetaDataImpl;
import cn.boat.router.consumer.PartitionData;
import cn.boat.router.consumer.SubscribeTopcisImpl;
import cn.boat.router.consumer.TopicWithPartitionData;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.util.ArrayList;
import java.util.List;

import com.bjtu.boatmq.common.NettyClient;
import com.bjtu.boatmq.common.protocol.Message;
import com.bjtu.boatmq.common.protocol.PacketSource;
import com.bjtu.boatmq.common.protocol.RequestCode;
import com.bjtu.boatmq.common.protocol.SourceDecoder;
import com.bjtu.boatmq.common.protocol.SourceEncoder;
import com.bjtu.boatmq.common.protocol.request.OffsetCommitRequest;
import com.bjtu.boatmq.common.protocol.PacketSource;
import com.bjtu.boatmq.common.protocol.request.PullMessageRequest;

/**
 * 使用此Consumer类
 * @author Dang
 * @create 2020-06-11 22:35
 */
public class DefaultMQConsumer {
    private String topicName;
    private GetConsumerMetaDataImpl getConsumerMetaData = new GetConsumerMetaDataImpl();
    private Consumer2Router consumer2Router;
    private NettyClient nettyClient;
    PacketSource packetsource=new PacketSource();
    List<Message> messageList;
    /**
     * 实例化一个Consumer对象
     * @param topicName 该Consumer类的topic
     */
    public DefaultMQConsumer(String topicName){
        consumer2Router = new Consumer2Router(topicName);
       // nettyClient = new NettyClient("",0);//先保证不报错，此处为Broker的ip地址和端口号以及通信包的对象
    }
    /**
     * 从broker拉取消息
     */
    public void pullMessage(String IP,Integer Port,String topicName,Integer partitionId) throws InterruptedException {
    	// 创建发送的数据包内容
        PullMessageRequest pullMessagePacket = new PullMessageRequest();

        pullMessagePacket.setTopicName(this.topicName);
        pullMessagePacket.setPartitionId(partitionId);
        pullMessagePacket.setConsumeOffset(PullMessageRequest.OFFSET_START);
       // 创建发送的数据包头部

        packetsource.setCode(RequestCode.PULL_MESSAGE);
        packetsource.setSponsorId("consumer2");
        packetsource.setSponsorType(2);
        packetsource.setFlag(0);
        packetsource.setData(pullMessagePacket);
        //通过Netty连接Broker,从Broker获得响应数据包
       // nettyClient.initialize();
        nettyClient=new NettyClient(IP, Port);
        nettyClient.send(packetsource);

        Bootstrap bootstrap = new Bootstrap();
        NioEventLoopGroup group = new NioEventLoopGroup();

        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel ch) {
                        ch.pipeline().addLast(new SourceEncoder());
                        ch.pipeline().addLast(new SourceDecoder());
                        ch.pipeline().addLast(new SimpleChannelInboundHandler<PacketSource>() {
                            @Override
                            protected void channelRead0(ChannelHandlerContext ctx, PacketSource ps) {
                                System.out.println("接收到一条broker返回的消息["+ps.getRequestId()+"]:"+ps.getRemark());
                                //System.out.println("offset:"+ps.getRemark());
                                messageList = (ArrayList<Message>)ps.getData();                                
                               // System.out.println("msg size:"+messageList.toString());
                            }
                        });
                    }
                });
        Integer offset = 0;
        for(Message message : messageList){
            //userCallback(message);
            System.out.println("消费消息：" + message);
            offset = message.getOffset();
        }
        commitOffset(topicName,partitionId,offset);
    }

    public void commitOffset(String topicName,Integer partitionId,Integer offset) throws InterruptedException{
        // 创建发送的数据包
        OffsetCommitRequest offsetCommitPacket = new OffsetCommitRequest();
        // 赋值数据包属性
        offsetCommitPacket.setTopicName(topicName);
        offsetCommitPacket.setPartitionId(partitionId);
        offsetCommitPacket.setHaveConsumedOffset(offset); // 最后一条消费的偏移量
        packetsource.setCode(RequestCode.PULL_MESSAGE);
        packetsource.setSponsorId("consumer001");
        packetsource.setSponsorType(2);
        packetsource.setFlag(0);
        packetsource.setData(offsetCommitPacket);
        // 调用Netty发送消息（尚待完善）
       // nettyClient.initialize(ipAddress,10110);
        nettyClient.send(packetsource);
    }
    /**
     * 启动线程不断地拉取消息
     */
    public void start() throws Exception {
    	
        consumer2Router.connectZookeeper(); //连接router
        consumer2Router.register();//注册节点
        ConsumerMetaData consumerMetaData = consumer2Router.subscribe();//订阅主题      
        new Thread(new Runnable() {
            @Override
            public void run() {
            	
                for(TopicWithPartitionData topicWithPartitionData : consumerMetaData.getTopicWithPartitionDataList()){
                    for(PartitionData partitionData : topicWithPartitionData.getPartitionDataList()){
                        //System.out.println(partitionData.getPartitionLeader());
                      try {
                    	System.out.println("开始拉取消息，PartitionLeader："+partitionData.getPartitionLeader());
						pullMessage(partitionData.getLeaderIp(),partitionData.getLeaderPort(),topicName,partitionData.getPartitionLeader());
						Thread.sleep(1000);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
                    }
                }
           }
        }).start();
    }


    public String getTopicName() {
        return topicName;
    }

    public void setTopicName(String topicName) {
        this.topicName = topicName;
    }
}
